//package org.example.stream;
//
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.data.redis.connection.RedisConnectionFactory;
//import org.springframework.data.redis.connection.stream.Consumer;
//import org.springframework.data.redis.connection.stream.ReadOffset;
//import org.springframework.data.redis.connection.stream.StreamOffset;
//import org.springframework.data.redis.stream.StreamMessageListenerContainer;
//import org.springframework.data.redis.stream.Subscription;
//import java.time.Duration;
//
///**
// * redis stream 配置（redis5.0以上）
// * https://www.zhihu.com/tardis/bd/art/440831491?source_id=1001
// * stream：每个Stream都有唯一的名称，它就是Redis的key；
// *   group：每个 Stream 都可以挂多个消费组；每个消费组 (Consumer Group) 的状态都是独立的，相互不受影响。也就是说同一份Stream 内部的消息会被每个消费组都消费到；
// *   consumer：同一个消费组 (Consumer Group) 可以挂接多个消费者 (Consumer)，这些消费者之间是竞争关系，即一条消息仅可被一个组内的一个消费者消费。每个消费者有一个组内唯一名称。不同组是可以全面消费的（重复消费），同一组内会发生竞争关系，只有其中一个可以消费。
// * ————————————————
// * 版权声明：本文为CSDN博主「喜欢你没道理_」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
// * 原文链接：https://blog.csdn.net/sinat_25849999/article/details/125797334
// *
// */
//@Configuration
//public class RedisStreamConfig {
//
//    @Autowired
//    private ListenerMessage listenerMessage;
//    @Autowired
//    private ListenerMessage1 listenerMessage1;
//    @Autowired
//    private ListenerMessage2 listenerMessage2;
//
//    @Bean
//    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ?> streamMessageListenerContainerOptions(){
//        return StreamMessageListenerContainer
//                        .StreamMessageListenerContainerOptions
//                        .builder()
//                        .pollTimeout(Duration.ofSeconds(1))
//                        .build();
//    }
//
//    @Bean
//    public StreamMessageListenerContainer streamMessageListenerContainer(RedisConnectionFactory factory,
//         StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ?> streamMessageListenerContainerOptions){
//        StreamMessageListenerContainer listenerContainer = StreamMessageListenerContainer.create(factory,
//                streamMessageListenerContainerOptions);
//        listenerContainer.start();
//        return listenerContainer;
//    }
//
//    /**XGROUP CREATE stream1 group1 0 MKSTREAM
//     * 订阅者1，消费组group1，收到消息后自动确认，与订阅者2为竞争关系，消息仅被其中一个消费
//     * @param streamMessageListenerContainer
//     * @return
//     */
//    @Bean
//    public Subscription subscription(StreamMessageListenerContainer streamMessageListenerContainer){
//        Subscription subscription = streamMessageListenerContainer.receiveAutoAck(
//                Consumer.from("group1","name1"),
//                StreamOffset.create("stream1", ReadOffset.lastConsumed()),
//                listenerMessage
//        );
//        return subscription;
//    }
//
//    /**
//     * 订阅者2，消费组group1，收到消息后自动确认，与订阅者1为竞争关系，消息仅被其中一个消费
//     * @param streamMessageListenerContainer
//     * @return
//     */
//    @Bean
//    public Subscription subscription2(StreamMessageListenerContainer streamMessageListenerContainer){
//        Subscription subscription = streamMessageListenerContainer.receiveAutoAck(
//                Consumer.from("group1","name2"),
//                StreamOffset.create("stream1", ReadOffset.lastConsumed()),
//                listenerMessage1
//        );
//        return subscription;
//    }
//
//    /**XGROUP CREATE stream1 group2  1
//     * 订阅者3，消费组group2，收到消息后不自动确认，需要用户选择合适的时机确认，与订阅者1和2非竞争关系，即使消息被订阅者1或2消费，亦可消费
//     *
//     * 当某个消息被ACK，PEL列表就会减少
//     * 如果忘记确认（ACK），则PEL列表会不断增长占用内存
//     * 如果服务器发生意外，重启连接后将再次收到PEL中的消息ID列表
//     * @param streamMessageListenerContainer
//     * @return
//     */
////    @Bean
//    public Subscription subscription3(StreamMessageListenerContainer streamMessageListenerContainer){
//        Subscription subscription = streamMessageListenerContainer.receive(
//                Consumer.from("group2","name1"),
//                StreamOffset.create("stream1", ReadOffset.lastConsumed()),
//                listenerMessage2
//        );
//        return subscription;
//    }
//
//}